package mq

import (
	"context"
	"fmt"
	"log"
	"strings"
	"sync"
	"task_queue/internal/models"
	"time"

	"github.com/go-redis/redis/v8"
	"gorm.io/gorm"
)

// RedisPlugin 实现基于 Redis 的消息队列
type RedisPlugin struct {
	client     *redis.Client
	closed     bool
	startTime  int64
	errorCount int
	lastError  string
	queueStats map[string]*QueueStats
	statsMutex sync.RWMutex
	db         *gorm.DB       // 添加数据库连接
	wg         sync.WaitGroup // 用于管理 goroutine 生命周期
}

// 确保 RedisPlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*RedisPlugin)(nil)

// newRedisPlugin 创建一个新的 RedisPlugin 实例
func newRedisPlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取 Redis 服务器地址
	addr, ok := config["addr"].(string)
	if !ok || addr == "" {
		return nil, fmt.Errorf("Redis 服务器地址配置缺失或无效")
	}

	// 可选的密码配置
	password := ""
	if pwd, ok := config["password"].(string); ok {
		password = pwd
	}

	// 可选的数据库索引配置
	db := 0
	if dbIndex, ok := config["db"].(int); ok {
		db = dbIndex
	}

	// 获取数据库连接
	var dbConn *gorm.DB
	if dbInterface, ok := config["db"]; ok {
		if conn, ok := dbInterface.(*gorm.DB); ok {
			dbConn = conn
		}
	}

	client := redis.NewClient(&redis.Options{
		Addr:     addr,
		Password: password,
		DB:       db,
	})

	// 测试连接
	ctx := context.Background()
	if _, err := client.Ping(ctx).Result(); err != nil {
		return nil, fmt.Errorf("连接到 Redis 服务器失败: %w", err)
	}

	return &RedisPlugin{
		client:     client,
		closed:     false,
		startTime:  time.Now().Unix(),
		queueStats: make(map[string]*QueueStats),
		db:         dbConn,
		wg:         sync.WaitGroup{},
	}, nil
}

// Type 返回消息队列类型
func (p *RedisPlugin) Type() MQType {
	return Redis
}

// Publish 发布消息到指定主题
func (p *RedisPlugin) Publish(ctx context.Context, topic string, message string) error {
	// 检查是否已关闭
	if p.closed {
		return fmt.Errorf("Redis 连接已关闭")
	}

	err := p.client.Publish(ctx, topic, message).Err()
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("向 Redis 发布消息失败: %v", err)
		return fmt.Errorf("向 Redis 发布消息失败: %w", err)
	}

	// 记录消息历史（如果有数据库连接）
	if p.db != nil {
		messageHistory := models.MessageHistory{
			Topic:     topic,
			MQType:    Redis,
			Message:   message,
			Status:    models.MessageStatusPending,
			CreatedAt: time.Now(),
		}

		// 异步保存，避免阻塞发布过程
		go func(history models.MessageHistory) {
			if err := p.db.Create(&history).Error; err != nil {
				fmt.Printf("记录消息历史失败: %v\n", err)
			}
		}(messageHistory)
	}

	return nil
}

// Subscribe 订阅指定主题的消息
func (p *RedisPlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	// 检查是否已关闭
	if p.closed {
		return fmt.Errorf("Redis 连接已关闭")
	}

	pubsub := p.client.Subscribe(ctx, topic)
	defer pubsub.Close()

	// 初始化或更新队列统计信息
	p.statsMutex.Lock()
	if _, exists := p.queueStats[topic]; !exists {
		p.queueStats[topic] = &QueueStats{
			TopicName: topic,
		}
	}
	stats := p.queueStats[topic]
	p.statsMutex.Unlock()

	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		ch := pubsub.Channel()
		for {
			select {
			case <-ctx.Done():
				return
			case msg, ok := <-ch:
				if !ok {
					return
				}

				// 更新统计信息
				p.statsMutex.Lock()
				stats.LastConsumedAt = time.Now().Unix()
				stats.ConsumedMessages++
				p.statsMutex.Unlock()

				err := handler(msg.Payload)
				if err != nil {
					log.Printf("处理消息失败: %v", err)

					// 更新失败统计
					p.statsMutex.Lock()
					stats.FailedMessages++
					p.statsMutex.Unlock()

					// 如果是因为达到消费数量限制而返回的错误，不应增加失败计数
					if !strings.Contains(err.Error(), "已达到指定消费数量") {
						// 更新消息历史为失败（如果有数据库连接）
						if p.db != nil {
							// 查找并更新消息状态
							go func(message string, errorMsg string) {
								// 查找最近的一条待处理消息
								var history models.MessageHistory
								if err := p.db.Where("topic = ? AND mq_type = ? AND message = ? AND status = ?",
									topic, Redis, message, models.MessageStatusPending).
									Order("created_at DESC").
									First(&history).Error; err == nil {

									// 更新为失败状态
									history.Status = models.MessageStatusFailed
									history.ErrorMessage = errorMsg
									p.db.Save(&history)
								}
							}(msg.Payload, err.Error())
						}
					} else {
						// 如果是因为达到消费数量限制，将消息放回队列
						go func(message string) {
							// 使用新的上下文，避免被取消
							newCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
							defer cancel()
							p.Publish(newCtx, topic, message)
						}(msg.Payload)
					}
				} else {
					// 更新消息历史为已消费（如果有数据库连接）
					if p.db != nil {
						now := time.Now()
						// 查找并更新消息状态
						go func(message string, consumedTime time.Time) {
							// 查找最近的一条待处理消息
							var history models.MessageHistory
							if err := p.db.Where("topic = ? AND mq_type = ? AND message = ? AND status = ?",
								topic, Redis, message, models.MessageStatusPending).
								Order("created_at DESC").
								First(&history).Error; err == nil {

								// 更新为已消费状态
								history.Status = models.MessageStatusConsumed
								history.ConsumedAt = &consumedTime
								p.db.Save(&history)
							}
						}(msg.Payload, now)
					}
				}
			}
		}
	}()

	return nil
}

// Close 关闭 Redis 连接
func (p *RedisPlugin) Close() error {
	if p.closed {
		return nil
	}
	p.closed = true
	p.wg.Wait() // 等待所有 goroutine 完成
	return p.client.Close()
}

// GetStatus 获取Redis的状态
func (p *RedisPlugin) GetStatus() MQStatus {
	// 检查连接状态
	ctx := context.Background()
	_, err := p.client.Ping(ctx).Result()
	isConnected := err == nil

	if !isConnected {
		p.errorCount++
		p.lastError = err.Error()
	}

	return MQStatus{
		IsConnected:    isConnected,
		ConnectionInfo: p.client.Options().Addr,
		StartTime:      p.startTime,
		ErrorCount:     p.errorCount,
		LastError:      p.lastError,
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (p *RedisPlugin) GetQueueStats(topic string) (QueueStats, error) {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := p.queueStats[topic]
	if !exists {
		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// Redis Pub/Sub 模式不存储消息，所以没有待处理消息数
	// 这里只返回已有的统计信息
	return *stats, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (p *RedisPlugin) ListQueueStats() []QueueStats {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(p.queueStats))
	for _, stats := range p.queueStats {
		result = append(result, *stats)
	}

	return result
}

// GetPendingMessages 获取待处理的消息
func (p *RedisPlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 检查主题是否存在
	p.statsMutex.RLock()
	_, exists := p.queueStats[topic]
	if !exists {
		p.statsMutex.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}
	p.statsMutex.RUnlock()

	// 如果没有数据库连接，返回空列表
	if p.db == nil {
		return []string{}, nil
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 从数据库查询待处理的消息
	var histories []models.MessageHistory
	if err := p.db.Where("topic = ? AND mq_type = ? AND status = ?",
		topic, Redis, models.MessageStatusPending).
		Order("created_at DESC").
		Offset(offset).
		Limit(pageSize).
		Find(&histories).Error; err != nil {
		return nil, err
	}

	// 提取消息内容
	messages := make([]string, len(histories))
	for i, history := range histories {
		messages[i] = history.Message
	}

	return messages, nil
}

// GetConsumedMessages 获取已消费的消息
func (p *RedisPlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 检查主题是否存在
	p.statsMutex.RLock()
	_, exists := p.queueStats[topic]
	if !exists {
		p.statsMutex.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}
	p.statsMutex.RUnlock()

	// 如果没有数据库连接，返回空列表
	if p.db == nil {
		return []MessageRecord{}, nil
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 从数据库查询已消费的消息
	var histories []models.MessageHistory
	if err := p.db.Where("topic = ? AND mq_type = ? AND status = ?",
		topic, Redis, models.MessageStatusConsumed).
		Order("consumed_at DESC").
		Offset(offset).
		Limit(pageSize).
		Find(&histories).Error; err != nil {
		return nil, err
	}

	// 转换为MessageRecord格式
	records := make([]MessageRecord, len(histories))
	for i, history := range histories {
		var timestamp int64
		if history.ConsumedAt != nil {
			timestamp = history.ConsumedAt.Unix()
		} else {
			timestamp = history.CreatedAt.Unix()
		}

		records[i] = MessageRecord{
			Message:   history.Message,
			Timestamp: timestamp,
		}
	}

	return records, nil
}

// GetFailedMessages 获取失败的消息
func (p *RedisPlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 检查主题是否存在
	p.statsMutex.RLock()
	_, exists := p.queueStats[topic]
	if !exists {
		p.statsMutex.RUnlock()
		return nil, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}
	p.statsMutex.RUnlock()

	// 如果没有数据库连接，返回空列表
	if p.db == nil {
		return []MessageRecord{}, nil
	}

	// 计算分页参数
	if page < 1 {
		page = 1
	}
	if pageSize < 1 {
		pageSize = 10
	}
	offset := (page - 1) * pageSize

	// 从数据库查询失败的消息
	var histories []models.MessageHistory
	if err := p.db.Where("topic = ? AND mq_type = ? AND status = ?",
		topic, Redis, models.MessageStatusFailed).
		Order("created_at DESC").
		Offset(offset).
		Limit(pageSize).
		Find(&histories).Error; err != nil {
		return nil, err
	}

	// 转换为MessageRecord格式
	records := make([]MessageRecord, len(histories))
	for i, history := range histories {
		records[i] = MessageRecord{
			Message:   history.Message,
			Timestamp: history.CreatedAt.Unix(),
			Error:     history.ErrorMessage,
		}
	}

	return records, nil
}

// CloseQueue 关闭指定主题的队列
func (p *RedisPlugin) CloseQueue(topic string) error {
	// Redis Pub/Sub 模式不需要显式关闭队列
	// 但我们可以从统计信息中移除该主题

	p.statsMutex.Lock()
	defer p.statsMutex.Unlock()

	// 检查主题是否存在
	_, exists := p.queueStats[topic]
	if !exists {
		return fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// 从统计信息中移除
	delete(p.queueStats, topic)

	return nil
}

// ListTopics 获取所有主题列表
func (p *RedisPlugin) ListTopics() []string {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	topics := make([]string, 0, len(p.queueStats))
	for topic := range p.queueStats {
		topics = append(topics, topic)
	}

	return topics
}
